/*
 * Copyright 2019 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.example.mqtt.heartBeat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public final class MqttHeartBeatClient {

    private static final Logger logger = LoggerFactory.getLogger(MqttHeartBeatClient.class);

    private String HOST;
    private int PORT;
    private String CLIENT_ID;
    private String USER_NAME;
    private String PASSWORD;

    private Bootstrap bootstrap;
    private Channel channel;


    public MqttHeartBeatClient(String HOST, int PORT, String CLIENT_ID, String USER_NAME, String PASSWORD) {
        this.HOST = HOST;
        this.PORT = PORT;
        this.CLIENT_ID = CLIENT_ID;
        this.USER_NAME = USER_NAME;
        this.PASSWORD = PASSWORD;
        initClient();
        connect();
    }

    /**
     * 初始化客户端
     */
    public void initClient() {
        bootstrap = new Bootstrap();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final MqttHeartBeatClient client = this;
        bootstrap.group(workerGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast("encoder", MqttEncoder.INSTANCE);
                ch.pipeline().addLast("decoder", new MqttDecoder());
                ch.pipeline().addLast("heartBeatHandler", new IdleStateHandler(0, 20, 0, TimeUnit.SECONDS));
                ch.pipeline().addLast("handler", new MqttHeartBeatClientHandler(CLIENT_ID, USER_NAME, PASSWORD, client));
            }
        });
    }

    /**
     * 与服务端建立连接
     * （注册重连机制）
     */
    public void connect() {
        ChannelFuture future = bootstrap.connect(HOST, PORT);
        if (channel != null && channel.isActive()) {
            return;
        }
        // auto reconnect
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture futureListener) throws Exception {
                if (futureListener.isSuccess()) {
                    channel = futureListener.channel();
                    logger.info("netty client connect [{}] success", PORT);
                } else {
                    logger.info("netty client connect [{}] failed try 10s later", PORT);
                    futureListener.channel().eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                connect();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }, 3L, TimeUnit.SECONDS);
                }
            }
        });

        this.channel = future.channel();
    }


    private static final String PROTOCOL_NAME_MQTT_3_1_1 = "MQTT";
    private static final int PROTOCOL_VERSION_MQTT_3_1_1 = 4;


    public void send(String msg) throws InterruptedException {
        System.out.println("################################ 自定义发送");
        MqttFixedHeader pingreqFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false,
                MqttQoS.AT_MOST_ONCE, false, 0);
        MqttFixedHeader connectFixedHeader =
                new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttConnectVariableHeader connectVariableHeader =
                new MqttConnectVariableHeader(PROTOCOL_NAME_MQTT_3_1_1, PROTOCOL_VERSION_MQTT_3_1_1, true, true, false,
                        0, false, false, 20);
        MqttConnectPayload connectPayload = new MqttConnectPayload(CLIENT_ID, "little-chen", msg, USER_NAME, PASSWORD);
        MqttConnectMessage connectMessage =
                new MqttConnectMessage(connectFixedHeader, connectVariableHeader, connectPayload);
        channel.writeAndFlush(connectMessage).sync();
    }

}
